package com.study.flink.scala.day05_state

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.scala._

object MapWithState {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.enableCheckpointing(5000)

    env.setStateBackend(new FsStateBackend("file:///D:\\IDEA\\flink-study\\dir\\day05\\backend"))

    val lines = env.socketTextStream("node02", 8888)

    val keyed: KeyedStream[(String, Int), Tuple] = lines.flatMap(_.split(" ")).map(w => {
      if ("error".equals(w)) {
        throw new RuntimeException("Error By Hand!")
      }
      (w, 1)
    }).keyBy(0)

    // 输入和历史数据
    val summed = keyed.mapWithState((input: (String, Int), state: Option[Int]) => {
      state match {
        case Some(count) => {
          val key = input._1
          val value = input._2
          val total = count + value
          ((key, total), Some(total))
        }
        case None => {
          (input, Some(input._2))
        }
      }
    })
    summed.print()
    env.execute("MapWithState-scala")
  }

}
